package com.miao.im.support.delayqueue;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/3/16 16:27
 * @Description: 延时消息管理器
 */
@Slf4j
public class DelayMessageManager {
    private final static int DEFAULT_THREAD_NUM = 5;
    private static int thread_num = DEFAULT_THREAD_NUM;
    private ExecutorService executor;
    /**任务主线程*/
    private Thread taskThread;
    /**延时队列*/
    private final DelayQueue<DelayMessageTask<?>> delayQueue;
    /**已执行的任务数量*/
    private static final AtomicLong EXECUTED_TASK_NUM = new AtomicLong(0);
    private static DelayMessageManager instance = new DelayMessageManager();

    private DelayMessageManager() {
        executor = Executors.newFixedThreadPool(thread_num);
        delayQueue = new DelayQueue<>();
        init();
    }

    public static DelayMessageManager getInstance() {
        return instance;
    }

    /**
     * 初始化
     */
    private void init() {
        taskThread = new Thread(this::execute);
        taskThread.setName("DelayQueueMonitor");
        taskThread.start();
    }


    private void execute() {
        while (true) {
            try {
                if (taskThread.isInterrupted()) {
                    return;
                }
                log.info("当前剩余延时任务数量:{}", delayQueue.size());
                DelayMessageTask<?> delayOrderTask = delayQueue.take();
                if (delayOrderTask.hasRemoved()) {
                    log.warn("任务已被移除，不需要重复执行");
                    continue;
                }
                Runnable task = delayOrderTask.getTask();
                if (null == task) {
                    continue;
                }
                // 提交到线程池执行task
                executor.execute(task);
                EXECUTED_TASK_NUM.addAndGet(1);
            } catch (Exception e) {
                log.error("DelayMessageManager:", e);
            }
        }
    }

    /**
     * 销毁
     * @author miaoguoxin
     * @date 2021/3/16
     */
    public void destroy() {
        delayQueue.clear();
        taskThread.interrupt();
        executor.shutdown();
        log.info("延迟消息管理器已被销毁");
    }

    /**
     * 添加任务
     */
    public void put(int id, String sessionId, Runnable task, long time, TimeUnit unit) {
        DelayMessageTask<?> delayOrder = new DelayMessageTask<>(id, sessionId, task, time, unit);
        delayQueue.put(delayOrder);
    }

    public void put(DelayMessageTask<?> delayOrder) {
        delayQueue.put(delayOrder);
    }

    /**
     * 删除任务
     * @param task
     * @return
     */
    public void removeTask(DelayMessageTask<?> task) {
        //防止多个重复对象的情况发生，只要相同的全部删除
        task.markRemoved();
        for (; ; ) {
            if (!delayQueue.remove(task)) {
                return;
            }
        }
    }
}
